package com.joysuch.wwyt.streampusher.publisher;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.joysuch.wwyt.streampusher.exception.PushException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;

/**
 * ffmpeg推流器
 *
 * @author qingzhou
 *         2017-11-16 17:28
 */
public class FFMpegRtmpStreamPusher implements StreamPusher {

    private static final Logger logger = LoggerFactory.getLogger(FFMpegRtmpStreamPusher.class);

    /**
     * 是否关闭
     */
    private volatile boolean closed;
    /**
     * 推流程序
     */
    private Process proc;
    /**
     * 流的标识，同一个原始流，应该使用同一个标识
     * 避免反复创建同一个流提高性能
     */
    private String identify;
    /**
     * 输入源
     */
    private String input;
    /**
     * 输出
     */
    private String output;
    /**
     * 视频宽度
     */
    private int imgWidth;
    /**
     * 视频高度
     */
    private int imgHeight;
    /**
     * 帧率
     */
    private int frameRate;

    /**
     * 是否使用TCP
     */
    private boolean tcp;
    /**
     * 是否转码
     */
    private boolean ffmpegTranscodeSwitch;
    /**
     * 执行取流操作的线程池
     */
    private ScheduledThreadPoolExecutor executor;

    /**
     * 所属应用名称
     */
    private String appName = "zqdwxt";

    private String ffmpegRoot = System.getProperty("ffmpeg.parent.path");

    /**
     *
     * @param identify
     * @param input
     * @param output
     * @param imgWidth
     * @param imgHeight
     * @param frameRate
     * @param tcp
     * @param ffmpegTranscodeSwitch 是否打开转码
     */
    public FFMpegRtmpStreamPusher(final String identify, String input, String output, int imgWidth, int imgHeight, int frameRate, boolean tcp, boolean ffmpegTranscodeSwitch) {
        this.closed = true;
        this.identify = identify;
        this.input = input;
        this.output = output;
        this.imgWidth = imgWidth;
        this.imgHeight = imgHeight;
        this.frameRate = frameRate;
        this.tcp = tcp;
        this.ffmpegTranscodeSwitch = ffmpegTranscodeSwitch;
        this.executor = new ScheduledThreadPoolExecutor(3, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread newT = new Thread(r);
                newT.setName("ffmpeg-publisher-" + identify);
                return newT;
            }
        });
    }

    @Override
    public String identify() {
        return this.identify;
    }

    @Override
    public boolean isClosed() {
        return closed;
    }

    @Override
    public void close() throws PushException {
        closed = true;
        if (proc != null) {
            try {
                proc.destroy();
            } catch (Exception e) {
                logger.error(identify + " destory process exception(ignored) ", e);
            }
        }
        this.executor.shutdown();
    }

    @Override
    public void open() throws PushException {
        String prefix = "/home/seekcy/softs/ffmpeg";
        if (!Strings.isNullOrEmpty(ffmpegRoot)) {
            prefix = ffmpegRoot;
        }
        List<String> cmd = Lists.newArrayList();

        cmd.add(prefix + "/ffmpeg");

        if(tcp){
            cmd.add("-rtsp_transport");
            cmd.add("tcp");
        }

        cmd.add("-i");
        cmd.add(this.input);

        cmd.add("-f");
        cmd.add("flv");

        cmd.add("-b:v");
        cmd.add("500k");

        if(this.ffmpegTranscodeSwitch){
            cmd.add("-vcodec");
            cmd.add("libx264");
        }else{
            cmd.add("-c");
            cmd.add("copy");
        }

        cmd.add("-r");
        cmd.add(String.valueOf(this.frameRate));

        cmd.add("-s");
        cmd.add(String.valueOf(this.imgWidth) + "*" + String.valueOf(this.imgHeight));

        cmd.add("-an");
        cmd.add(this.output + (this.appName == null ? "" : ("_" + this.appName)));

        printCmd(cmd);
        try {
            ProcessBuilder builder = new ProcessBuilder(cmd);
            builder.redirectErrorStream(true);
            proc = builder.start();
            this.executor.execute(new ProcExitHandler(proc, this));
            closed = false;
        } catch (IOException e) {
            throw new PushException("执行ffmpeg出现异常", e.getCause());
        }
    }

    private void printCmd(List<String> cmd) {
        StringBuilder buf = new StringBuilder();
        for (String c : cmd) {
            buf.append(c).append(" ");
        }
        logger.info("执行推流命令:" + buf.toString());
    }

    public class ProcExitHandler implements Runnable {

        private Process proc;
        private StreamPusher pusher;

        public ProcExitHandler(Process proc, StreamPusher pusher) {
            this.proc = proc;
            this.pusher = pusher;
        }

        @Override
        public void run() {
            try {
                wait(proc);
            } catch (InterruptedException e) {
                logger.error("===== exec ffmeg cmd InterruptedException error ", e);
            } catch (Exception e) {
                logger.error("===== exec ffmeg cmd error ", e);
            }
            try {
                logger.info("********** proc auto close " + input);
                pusher.close();
            } catch (PushException e) {
                e.printStackTrace();
            }
        }

        public int wait(Process exec) throws Exception {
            logger.info("********** proc.waitfor() " + input);
            //读取标准错误流
            final BufferedReader brError = new BufferedReader(new InputStreamReader(exec.getErrorStream(), "utf8"));
            //读取标准输入流
            final BufferedReader br = new BufferedReader(new InputStreamReader(exec.getInputStream()));
            Thread infoInputThread = new Thread() {
                public void run() {
                    try {
                        while ((brError.readLine()) != null) {

                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            if (brError != null)
                                brError.close();
                        } catch (IOException e) {
                            logger.error("===== read ffmeg cmd errorput error ", e);
                        }
                    }
                }
            };
            executor.execute(infoInputThread);
            Thread errorInputThread = new Thread() {
                public void run() {
                    try {
                        while ((br.readLine()) != null) {

                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    } finally {
                        try {
                            if (br != null)
                                br.close();
                        } catch (IOException e) {
                            logger.error("===== read ffmeg cmd input error ", e);
                        }
                    }
                }
            };
            executor.execute(errorInputThread);
            return exec.waitFor();
        }
    }

    public String getAppName() {
        return appName;
    }

    public void setAppName(String appName) {
        this.appName = appName;
    }
}
